package org.example;

import io.netty.util.internal.StringUtil;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.common.StringUtils;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public class ZkDistributedLock {

    public static final String ROOT_PATH = "/distribute";
    private String path;
    private final ZooKeeper zooKeeper;

    public static final ThreadLocal<Integer> THREAD_LOCAL = new ThreadLocal<>();

    public ZkDistributedLock(ZooKeeper zooKeeper, String lockName){
        this.zooKeeper = zooKeeper;
        try {
            this.path = zooKeeper.create(ROOT_PATH + "/" + lockName + "_",
                    null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void lock() throws InterruptedException {
        Integer flag = THREAD_LOCAL.get();
        if (flag != null && flag > 0) {
            THREAD_LOCAL.set(flag + 1);
            return;
        }
        String preNode = getpreNode(path);
        //如果该节点没有前一个节点，说明该节点是最小的节点
        if (null == preNode) {
            THREAD_LOCAL.set(1);
        } else {
            System.out.println("存在前置节点:" + preNode);
            CountDownLatch countDownLatch = new CountDownLatch(1);
            try {
                if (this.zooKeeper.exists(ROOT_PATH + "/" + preNode, watchedEvent -> {
                    System.out.println("通知成功");
                    countDownLatch.countDown();
                }) == null) {
                    return;
                }
                System.out.println("等待通知中");
                countDownLatch.await();
                THREAD_LOCAL.set(1);
                return;

            } catch (KeeperException | InterruptedException e) {
                e.printStackTrace();
            }
            try {
                System.out.println("休息下下");
                Thread.sleep(200);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            lock();
        }
    }

    private String getpreNode(String path) {
        //获取当前节点的序列化序号
        Long curSerial = Long.valueOf(StringUtil.substringAfter(path, '_'));
        //获取根路径下的所有序列化子节点
        try {
            List<String> nodes = this.zooKeeper.getChildren(ROOT_PATH, false);
            //判空处理
            if (nodes.isEmpty()) {
                return null;
            }
            //获取前一个节点
            Long flag = 0L;
            String preNode = null;
            for (String node : nodes) {
                //获取每个节点的序列化号
                Long serial = Long.valueOf(StringUtil.substringAfter(node, '_'));
                if (serial < curSerial && serial > flag) {
                    flag = serial;
                    preNode = node;
                }
            }
            return preNode;
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return null;
    }

    public void unlock(){
        try {
            THREAD_LOCAL.set(THREAD_LOCAL.get() - 1);
            if (THREAD_LOCAL.get() == 0) {
                this.zooKeeper.delete(path, 0);
                THREAD_LOCAL.remove();
            }

        } catch (InterruptedException | KeeperException e) {
            e.printStackTrace();
        }
    }
}
